#include "UdpSession.h"
#include "UdpDataPacket.h"

using namespace clusnet::net;

UdpSession::UdpSession(boost::asio::io_context& io, boost::asio::io_context::strand& strand, boost::asio::ip::udp::socket& socket, boost::asio::ip::udp::endpoint const& remoteEndpoint, UdpWriteCompleteCallback const& writeCallback)
    : io_(io),
      strand_(strand),
      socket_(socket),
      remoteEndpoint_(remoteEndpoint),
      writeCallback_(writeCallback)
{
}

UdpSession::~UdpSession()
{
    if (!sendMsgs_.empty())
    {
        LOG(ERROR) << sendMsgs_.size() << " message will be discarded from " << localEndpoint().address() << ":" << localEndpoint().port() << " to " << remoteEndpoint_.address() << ":" << remoteEndpoint_.port();
    }
}

void UdpSession::safeAsyncWrite(UdpDataPacketPtr const& data)
{
    auto self(shared_from_this());
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, self, data]()
    {
        bool write_in_progress = !sendMsgs_.empty();
        sendMsgs_.emplace_back(data);
        if (!write_in_progress)
        {
            asyncWrite();
        }
    }));
}

void UdpSession::asyncWrite()
{
    auto self(shared_from_this());
    socket_.async_send_to(boost::asio::buffer(sendMsgs_.front()->cdata(), sendMsgs_.front()->dataLength()), remoteEndpoint_, boost::asio::bind_executor(strand_, [this, self](boost::system::error_code const& ec, std::size_t length)
    {
        if (ec)
        {
            LOG(WARNING) << "async_write failed. " << ec.message();
            return;
        }
        if (length != sendMsgs_.front()->dataLength())
        {
            LOG(WARNING) << "async_write failed. data length not equal. expected send len: " << sendMsgs_.front()->dataLength() << " actual send len: " << length;
            return;
        }

        if (writeCallback_)
        {
            auto self(shared_from_this());
            writeCallback_(self, sendMsgs_.front());
        }

        sendMsgs_.pop_front();
        if (!sendMsgs_.empty())
        {
            asyncWrite();
        }
    }));
}
